package sparkstreaming.lesson06

import org.apache.spark.streaming.dstream.ReceiverInputDStream
import org.apache.spark.{HashPartitioner, SparkConf, SparkContext}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
  * Created by Administrator on 2018/5/12.
  *
  * 需求：
  *   实现一个 每隔4秒，统计最近6秒的单词计数的情况。
  *
  *   reduceByKeyAndWindow
  */
object WindowOperatorTest {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[2]").setAppName("NetWordCount")
    val sc = new SparkContext(conf)
    val ssc = new StreamingContext(sc,Seconds(2))
    /**
      * 数据的输入
      * 到目前为止这个地方还没有跟生产进行对接。
      */
    val dstream: ReceiverInputDStream[String] = ssc.socketTextStream("hadoop1",9999)
    /**
      * 数据的处理
      * 我们一直讲的是数据处理的算子
      * 这个地方算子 就是生产时候用的算子。
      *
      *  reduceFunc: (V, V) => V,
         *windowDuration: Duration,6 窗口的大小
         *slideDuration: Duration,4  滑动的大小
      *numPartitions: Int  指定分区数
      */
    val resultWordCountDStream = dstream.flatMap(_.split(","))
      .map((_, 1))
      .reduceByKeyAndWindow((x: Int, y: Int) => x + y, Seconds(6), Seconds(4))

    /**
      * 数据的输出
      * 这个地方，我们还是没跟生产对接起来，我们下一节课去讲这个事。
      */
    resultWordCountDStream.print()
    /**
      * 这个操作仅仅限于测试的时候使用。
      */
    ssc.start()
    ssc.awaitTermination()
    ssc.stop()

  }

}
